package com.hliushi.spark.exmaple

import org.apache.spark.{SparkConf, SparkContext}

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/16 7:12
 */
object GroupTopN {

  /**
   * 使用Spark的RDD算子来处理这个 分组topN问题
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[6]").setAppName("group_topN")
    val sc = new SparkContext(conf)

    val source = sc.textFile("dataset/Order.csv")

    // 使用RDD处理数据
    // 切分数据, 提取关键字段
    // 根据type类型分组
    // 记性利润排序
    /**
     * 一个实用小技巧
     * .    写链式rdd算子运算编程的时候, 先把最终返回的变量写出来
     * .    然后可以看出每一步rdd算子运算的返回结果. 有助于下面的算子转换运算
     */
    val result = source.map((x: String) => {
      val arr = x.split(",")
      val bean = OrderBean(arr(0), arr(1), arr(2), arr(3), arr(4), arr(5).toDouble, arr(6).toInt, arr(7).toDouble, arr(8).toDouble)
      bean
    }).groupBy((bean: OrderBean) => bean.proType)
      .map((x: (String, Iterable[OrderBean])) => {
        val list = x._2.toList

        val beans = list.sortBy((bean: OrderBean) => bean.profit)
          .reverse
          .take(3)
        beans
      })

    result.foreach((x: List[OrderBean]) => {
      x.foreach((y: OrderBean) => y.print())
    })

    // 回收资源停止sc,结束任务
    sc.stop()
  }


  /**
   * 订单数据格式为:
   * 订单ID	订单日期	省/自治区 产品ID 类别  销售额	数量	折扣	利润
   *
   * @param orderId   订单ID
   * @param orderDate 订单日期
   * @param province  省/自治区
   * @param proId     产品ID
   * @param proType   类别
   * @param sales     销售额
   * @param total     数量
   * @param discount  折扣
   * @param profit    利润
   */
  case class OrderBean(var orderId: String,
                       var orderDate: String,
                       var province: String,
                       var proId: String,
                       var proType: String,
                       var sales: Double,
                       var total: Int,
                       var discount: Double,
                       var profit: Double) {
    def printResult(): String = {
      println(orderDate, proId, proType, total, profit)
      orderDate + " " + proId + " " + proType + " " + total + " " + profit
    }

    def print(): Unit = {
      println(orderDate, proId, proType, total, profit)
    }
  }

}